#ifndef __MYSERVER_H_
#define __MYSERVER_H_
#include <vector>
#include <iostream>
#include <assert.h>
#include <cstring>
#include <ctime>
#include <functional>
#include <unordered_map>
#include <thread>
#include <mutex>
#include <condition_variable>
#include <memory>
#include <typeinfo>
#include <fcntl.h>
#include <signal.h>
#include <unistd.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/epoll.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>

#define DBG_LEVEL 0
#define ERR_LEVEL 1
#define INF_LEVEL 2
#define DEF_LEVEL DBG_LEVEL
#define LOG(lev_str, level, format, ...)                                                              \
    {                                                                                                 \
        if (level >= DEF_LEVEL)                                                                       \
        {                                                                                             \
            time_t t = time(nullptr);                                                                 \
            struct tm *ptr = localtime(&t);                                                           \
            char t_str[32];                                                                           \
            strftime(t_str, 31, "%H:%M:%S", ptr);                                                     \
            printf("[%p][%s][%s][%s:%d]" format "\n", (void*)pthread_self(),lev_str, t_str, __FILE__, __LINE__, ##__VA_ARGS__); \
        }                                                                                             \
    }
#define DLOG(format, ...) LOG("DBG", DBG_LEVEL, format, ##__VA_ARGS__)
#define ELOG(format, ...) LOG("ERR", ERR_LEVEL, format, ##__VA_ARGS__)
#define ILOG(format, ...) LOG("INF", INF_LEVEL, format, ##__VA_ARGS__)
#define CAPCITY 1024

class Buffer
{
public:
    Buffer() : _write_idx(0), _read_idx(0), _buffer(CAPCITY) {}
    char *Begin()
    {
        return &*_buffer.begin();
    }
    // 获取当前写入起始地址, _buffer的空间起始地址，加上写偏移量
    char *WritePosition()
    {
        return Begin() + _write_idx;
    }
    // 获取当前读取起始地址
    char *ReadPosition()
    {
        return Begin() + _read_idx;
    }
    // 获取缓冲区末尾空闲空间大小--写偏移之后的空闲空间, 总体空间大小减去写偏移
    uint64_t TailIdleSize()
    {
        return _buffer.size() - _write_idx;
    }
    // 获取缓冲区起始空闲空间大小--读偏移之前的空闲空间
    uint64_t HeadIdleSize()
    {
        return _read_idx;
    }
    // 获取可读数据大小 = 写偏移 - 读偏移
    uint64_t ReadAbleSize()
    {
        return _write_idx - _read_idx;
    }
    // 将读偏移向后移动
    void MoveReadOffset(uint64_t len)
    {
        // 需要注意移动的大小要小于可读数据的大小
        assert(len <= ReadAbleSize());
        _read_idx += len;
    }
    // 将写偏移向后移动
    void MoveWriteOffset(uint64_t len)
    {
        assert(len <= TailIdleSize());
        _write_idx += len;
    }
    // 确保可写空间足够（整体空闲空间够了就移动数据，否则就扩容）
    void EnsureWriteSpace(uint64_t len)
    {
        if (len <= TailIdleSize())
            return;
        if (len <= TailIdleSize() + HeadIdleSize())
        {
            // 将当前可读数据大小保存
            uint64_t ret = ReadAbleSize();
            // 将数据全部移动到0位置起始
            std::copy(ReadPosition(), ReadPosition() + ret, Begin());
            _read_idx = 0;
            _write_idx = ret;
        }
        else
        {
            _buffer.resize(_write_idx + len);
        }
    }
    // 写入数据
    void Write(const void *data, uint64_t len)
    {
        if (len == 0)
            return;
        // 先看看空间够不够
        EnsureWriteSpace(len);
        // 再将数据写入到缓冲区
        const char *ret = (const char *)data;
        std::copy(ret, ret + len, WritePosition());
    }
    void WriteAndPush(const void *data, uint64_t len)
    {
        Write(data, len);
        MoveWriteOffset(len);
    }
    void WriteString(const std::string &data)
    {
        Write(data.c_str(), data.size());
    }
    void WriteStringAndPush(const std::string &data)
    {
        WriteString(data);
        MoveWriteOffset(data.size());
    }
    void WriteBuffer(Buffer &buff)
    {
        Write(buff.ReadPosition(), buff.ReadAbleSize());
    }
    void WriteBufferAndPush(Buffer &buff)
    {
        WriteBuffer(buff);
        MoveWriteOffset(buff.ReadAbleSize());
    }
    // 读取数据
    void Read(void *buf, uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(), ReadPosition() + len, (char *)buf);
    }
    void ReadAndPop(void *buf, uint64_t len)
    {
        Read(buf, len);
        MoveReadOffset(len);
    }
    std::string ReadString(uint64_t len)
    {
        assert(len <= ReadAbleSize());
        std::string ret;
        ret.resize(len);
        Read(&ret[0], len);
        // std::cout<<ret<<std::endl;
        return ret;
    }
    std::string ReadStringAndPop(uint64_t len)
    {
        std::string ret = ReadString(len);
        // std::cout<<ret<<std::endl;
        MoveReadOffset(len);
        return ret;
    }
    // 获取一行数据，以http协议为标准
    char *GetCRLF()
    {
        char *pos = (char *)memchr(ReadPosition(), '\n', ReadAbleSize());
        return pos;
    }
    std::string GetLine()
    {
        char *pos = GetCRLF();
        if (pos == nullptr)
        {
            return "";
        }
        return ReadString(pos - ReadPosition() + 1);
    }
    std::string GetLineAndPop()
    {
        std::string ret = GetLine();
        MoveReadOffset(ret.size());
        return ret;
    }
    // 清空缓冲区
    void Clear()
    {
        _read_idx = 0;
        _write_idx = 0;
    }

private:
    std::vector<char> _buffer;
    uint64_t _write_idx;
    uint64_t _read_idx;
};

#define _LIS_MAX 1024
class Socket
{
public:
    Socket() : _sockfd(-1) {}
    Socket(int fd) : _sockfd(fd) { NonBlock(); }
    ~Socket() { Close(); }
    // 创建套接字
    bool Create()
    {
        _sockfd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP);
        if (_sockfd < 0)
        {
            ELOG("create socket error");
            return false;
        }
        // NonBlock();
        return true;
    }
    // 绑定地址
    bool Bind(const std::string &ip, uint16_t port)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = bind(_sockfd, (struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ELOG("bind error");
            return false;
        }
        return true;
    }
    // 监听套接字
    bool Listen()
    {
        int ret = listen(_sockfd, _LIS_MAX);
        if (ret < 0)
        {
            ELOG("listen error");
            return false;
        }
        return true;
    }
    // 向服务器发起连接
    bool Connect(uint16_t port, const std::string &ip)
    {
        struct sockaddr_in addr;
        addr.sin_family = AF_INET;
        addr.sin_port = htons(port);
        addr.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len = sizeof(struct sockaddr_in);
        int ret = connect(_sockfd, (struct sockaddr *)&addr, len);
        if (ret < 0)
        {
            ELOG("connect error");
            return false;
        }
        return true;
    }
    // 获取新连接
    int Accept()
    {
        int newfd = accept(_sockfd, nullptr, nullptr);
        if (newfd < 0)
        {
            ELOG("accept error");
            return -1;
        }
        // // 设置新连接为非阻塞
        // int flags = fcntl(newfd, F_GETFL, 0);
        // fcntl(newfd, F_SETFL, flags | O_NONBLOCK);
        //DLOG("创建新连接成功 %d",newfd);
        return newfd;
    }
    // 接收数据
    ssize_t Recv(void *buf, size_t len, int flag = 0)
    {
        ssize_t n = recv(_sockfd, buf, len, flag);
        if (n < 0)
        {
            // EAGAIN表示当前socket缓冲区没有数据
            // EINTR表示被信号终止
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ELOG("recv error: %s", strerror(errno));
            return -1;
        }
        else if(n==0)
        {
            return 0;
        }
        return n;
    }
    // 发送数据
    ssize_t Send(const void *buf, size_t len, int flag = 0)
    {
        ssize_t n = send(_sockfd, buf, len, flag);
        if (n < 0)
        {
            if (errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            ELOG("send error");
            return -1;
        }
        return n;
    }
    // 读非阻塞
    ssize_t RecvNonBlock(void *buf, size_t len)
    {
        return Recv(buf, len, MSG_DONTWAIT);
    }
    ssize_t NonBlockSend(void *buf, size_t len)
    {
        if (len == 0)
            return 0;
        return Send(buf, len, MSG_DONTWAIT); // MSG_DONTWAIT 表示当前发送为非阻塞。
    }
    // 创建客户端连接
    bool CreateClient(uint16_t port, const std::string &ip)
    {
        if (Create() == false)
            return false;
        if (Connect(port, ip) == false)
            return false;
        DLOG("客户端连接创建成功");
        return true;
    }
    // 创建服务器连接
    bool CreateServer(uint16_t port, const std::string &ip = "0.0.0.0", bool flag = false)
    {
        if (Create() == false)
            return false;
        if (Bind(ip, port) == false)
            return false;
        if (Listen() == false)
            return false;
        if (flag)
            NonBlock();
        ReuseAddress();
        DLOG("服务器连接创建成功");
        return true;
    }
    // 关闭套接字
    void Close()
    {
        if (_sockfd > 0)
            close(_sockfd);
        _sockfd = -1;
    }
    // 设置端口可重用
    void ReuseAddress()
    {
        int ret = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEADDR, (void *)&ret, sizeof(int));
        ret = 1;
        setsockopt(_sockfd, SOL_SOCKET, SO_REUSEPORT, (void *)&ret, sizeof(int));
    }
    // 设置套接字属性为非阻塞
    void NonBlock()
    {
        int flag = fcntl(_sockfd, F_GETFL);
        fcntl(_sockfd, F_SETFL, flag | O_NONBLOCK);
    }
    int Fd() { return _sockfd; }

private:
    int _sockfd;
};

class EventLoop;
class Channel
{
public:
    using EventCallBack = std::function<void()>;
    Channel(int fd, EventLoop *eventloop) : _fd(fd), _eventloop(eventloop), _events(0), _revents(0) {}
    void SetReadCallBack(const EventCallBack &cb) { _read_callback = cb; };
    void SetWriteCallBack(const EventCallBack &cb) { _write_callback = cb; };
    void SetErrorCallBack(const EventCallBack &cb) { _error_callback = cb; };
    void SetCloseCallBack(const EventCallBack &cb) { _close_callback = cb; };
    void SetEventCallBack(const EventCallBack &cb) { _event_callback = cb; };
    int Fd() { return _fd; }
    uint32_t GetEvents() { return _events; }
    void SetRevents(uint32_t events)
    {
        _revents = events;
    }
    // 判断当前事件是否可读
    bool ReadAble()
    {
        return _events & EPOLLIN;
    }
    // 判断当前事件是否可写
    bool WriteAble()
    {
        return _events & EPOLLOUT;
    }
    // 启动可读事件监控
    void EnableRead()
    {
        _events |= EPOLLIN;
        Update();
    }
    // 启动可写事件监控
    void EnableWrite()
    {
        _events |= EPOLLOUT;
        Update();
    }
    // 关闭可读事件监控
    void DisableRead()
    {
        _events &= ~EPOLLIN;
        Update();
    }
    // 关闭可写事件监控
    void DisableWrite()
    {
        _events &= ~EPOLLOUT;
        Update();
    }
    // 关闭所有事件监控
    void DisableAll()
    {
        _events = 0;
        Update();
    }
    // 移除事件监控
    void Remove();
    void Update();
    // 当某个事件触发就调用这个处理函数
    void HandleEvent()
    {
        // EPOLLRDHUP 仅表示对端关闭了​​写方向​​（连接半关闭），此时本端仍可发送数据
        // EPOLLPRI 紧急数据，信号
        if ((_revents & EPOLLIN) || (_revents & EPOLLRDHUP) || (_revents & EPOLLPRI))
        {
            if (_read_callback)
                _read_callback();
        }
        /*有可能会释放连接的操作事件，一次只处理一个*/
        if (_revents & EPOLLOUT)
        {
            if (_write_callback)
                _write_callback();
        }
        else if (_revents & EPOLLERR)
        {
            if (_error_callback)
                _error_callback(); // 一旦出错，就会释放连接，因此要放到前边调用任意回调
        }
        else if (_revents & EPOLLHUP)
        {
            if (_close_callback)
                _close_callback();
        }
        /*不管任何事件，都调用的回调函数*/
        if (_event_callback)
            _event_callback();
    }
    // Poller *Poll() { return _poll; }

private:
    int _fd;
    uint32_t _events;  // 需要监控的事件
    uint32_t _revents; // 当前连续触发的事件
    EventLoop *_eventloop;
    EventCallBack _read_callback;  // 可读事件回调
    EventCallBack _write_callback; // 可写事件回调
    EventCallBack _error_callback; // 错误事件回调
    EventCallBack _close_callback; // 连接断开事件回调
    EventCallBack _event_callback; // 其他事件回调
};

#define _MAX_EPOLLEVENTS 1024
class Poller
{
public:
    Poller()
    {
        _epfd = epoll_create(_MAX_EPOLLEVENTS);
        if (_epfd < 0)
        {
            ELOG("epolll_create error");
            abort();
        }
    }
    // 修改事件监控
    void UpdateEvent(Channel *channel)
    {
        bool ret = HasChannel(channel);
        if (ret == false)
        {
            // 不存在则添加
            _channels.insert(std::make_pair(channel->Fd(), channel));
            return Update(channel, EPOLL_CTL_ADD);
        }
        // 存在即修改
        return Update(channel, EPOLL_CTL_MOD);
    }
    // 移除事件监控
    void RemoveEvent(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if (it != _channels.end())
        {
            _channels.erase(channel->Fd());
        }
        return Update(channel, EPOLL_CTL_DEL);
    }
    // 开启事件监控，返回活跃事件
    void Poll(std::vector<Channel *> &actives)
    {
        int ept = epoll_wait(_epfd, _evs, _MAX_EPOLLEVENTS, -1);
        if (ept < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ELOG("epoll_wait error:%s\n", strerror(errno));
            abort(); // 退出程序
        }
        for (int i = 0; i < ept; i++)
        {
            // DLOG("获取到一个活跃事件");
            auto it = _channels.find(_evs[i].data.fd);
            assert(it != _channels.end());
            it->second->SetRevents(_evs[i].events);
            actives.push_back(it->second);
        }
        return;
    }
    // 将要关闭的channel添加到待删除的表中，防止段错误
    void AppendDeletedChannel(Channel *channel)
    {
        _deleted_channels.push_back(channel);
    }
    void ClearDeletedChannels()
    {
        for (auto channel : _deleted_channels)
        {
            delete channel;
        }
        _deleted_channels.clear();
    }

private:
    void Update(Channel *channel, int op)
    {
        int fd = channel->Fd();
        struct epoll_event ev;
        ev.data.fd = fd;
        ev.events = channel->GetEvents();
        int ret = epoll_ctl(_epfd, op, fd, &ev);
        if (ret < 0)
        {
            DLOG("epoll_ctl error");
        }
        return;
    }
    // 判断channel是否已经添加到事件监控
    bool HasChannel(Channel *channel)
    {
        auto it = _channels.find(channel->Fd());
        if (it != _channels.end())
        {
            return true;
        }
        return false;
    }

private:
    int _epfd;
    struct epoll_event _evs[_MAX_EPOLLEVENTS];
    std::unordered_map<int, Channel *> _channels; // 描述符与channel的映射关系
    std::vector<Channel *> _deleted_channels;     // 待删除的 Channel 列表
};

using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimeTask
{
public:
    TimeTask(uint64_t id, uint32_t timeout, const TaskFunc &task) : _id(id), _timeout(timeout), _task(task), _check(false)
    {
    }
    ~TimeTask()
    {
        if (_check == false)
            _task(); // 当没有被取消时被执行
        _release();
    }
    void SetRelease(const ReleaseFunc &cb) { _release = cb; }
    void Cancel() { _check = true; }
    uint32_t TimeOut() { return _timeout; }
    const TaskFunc &GetTask() const { return _task; }

private:
    uint64_t _id;
    uint32_t _timeout; // 超时时间
    TaskFunc _task;
    ReleaseFunc _release;
    bool _check;
};
// 定时器
class TimerWheel
{
public:
    using PtrFunc = std::shared_ptr<TimeTask>;
    using WeakFunc = std::weak_ptr<TimeTask>;
    TimerWheel(EventLoop *loop) : _tick(0),
                                  _capcity(60),
                                  _wheel(_capcity),
                                  _loop(loop),
                                  _timefd(CreateTimeFd()),
                                  _channel(std::make_unique<Channel>(_timefd, _loop))
    {
        _channel->SetReadCallBack(std::bind(&TimerWheel::OnTime, this));
        _channel->EnableRead();
    }
    // 因为在添加定时任务时可能是多线程并发执行的，直接将该操作归于同一个线程来执行可以提高效率---EventLoop线程
    void TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &task);
    // 刷新/延长超时时间
    void TimerRefresh(uint64_t id);
    // 取消
    void TimerCancel(uint64_t id);
    bool HasTimer(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return false;
        }
        return true;
    }

private:
    // 这个函数每次执行一次，_tick就加1
    void RunTimerTask()
    {
        _tick = (_tick + 1) % _capcity;
        _wheel[_tick].clear();
    }
    int ReadTimeFd()
    {
        int64_t index = 0;
        ssize_t n = read(_timefd, &index, sizeof(index));
        if (n < 0)
        {
            ELOG("read timefd error");
            abort();
        }
        return index;
    }
    void OnTime()
    {
        // ReadTimeFd();
        // RunTimerTask();
        int n = ReadTimeFd();
        for (int i = 0; i < n; i++)
        {
            RunTimerTask();
        }
    }
    // 添加定时任务
    void TimerAddInLoop(uint64_t id, uint32_t timeout, const TaskFunc &task)
    {
        // 先构建一个对象
        PtrFunc pt(new TimeTask(id, timeout, task));
        pt->SetRelease(std::bind(&TimerWheel::RemoveTimers, this, id));
        // 将该任务添加到wheel
        _wheel[(pt->TimeOut() + _tick) % _capcity].push_back(pt);
        _timers[id] = WeakFunc(pt); // 直接构建一个weak_ptr防止循环引用
    }
    void RemoveTimers(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it != _timers.end())
        {
            _timers.erase(id);
        }
    }
    static int CreateTimeFd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC, 0);
        if (timerfd < 0)
        {
            ELOG("TIMERFD CREATE FAILED!");
            abort();
        }
        // int timerfd_settime(int fd, int flags, struct itimerspec *new, struct itimerspec *old);
        struct itimerspec itime;
        itime.it_value.tv_sec = 1;
        itime.it_value.tv_nsec = 0; // 第一次超时时间为1s后
        itime.it_interval.tv_sec = 1;
        itime.it_interval.tv_nsec = 0; // 第一次超时后，每次超时的间隔时
        timerfd_settime(timerfd, 0, &itime, NULL);
        return timerfd;
    }
    // 刷新
    void TimerRefreshInLoop(uint64_t id)
    {
        // 通过保存的WeakFunc对象构造一个PtrFunc对象
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return;
        }
        PtrFunc pt = it->second.lock();
        if (!pt)
            return;

        // 先取消旧任务
        pt->Cancel();

        // 创建新任务实例
        PtrFunc new_pt(new TimeTask(id, pt->TimeOut(), pt->GetTask()));
        new_pt->SetRelease(std::bind(&TimerWheel::RemoveTimers, this, id));

        // 添加到时间轮
        int slot = (new_pt->TimeOut() + _tick) % _capcity;
        _wheel[slot].push_back(new_pt);

        // 更新定时器映射
        _timers[id] = new_pt;
    }
    // 取消定时任务
    void TimerCancelInLoop(uint64_t id)
    {
        auto it = _timers.find(id);
        if (it == _timers.end())
        {
            return;
        }
        PtrFunc pt = it->second.lock();
        if (pt)
            pt->Cancel();
    }

private:
    int _tick;    // 当前时间轮的指向
    int _capcity; // 最大超时时间
    std::vector<std::vector<PtrFunc>> _wheel;
    std::unordered_map<uint64_t, WeakFunc> _timers; // 用来存储任务id与任务的映射关系，这样在时钟指到该任务时就能快速找到该任务

    EventLoop *_loop;
    int _timefd;
    std::unique_ptr<Channel> _channel;
};

class EventLoop
{
private:
    static int CreateEventFd()
    {
        int efd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (efd < 0)
        {
            ELOG("eventfd create error");
            abort();
        }
        return efd;
    }
    void ReadEventFd()
    {
        int64_t ret;
        int n = read(_eventfd, &ret, sizeof(ret));
        if (n <= 0)
        {
            if (errno == EINTR || errno == EAGAIN)
            {
                return;
            }
            ELOG("read eventfd error");
            abort();
        }
        return;
    }
    void WeakUpEventFd()
    {
        int64_t ret = 1;
        int n = write(_eventfd, &ret, sizeof(ret));
        if (n < 0)
        {
            if (errno == EINTR)
            {
                return;
            }
            ELOG("write eventfd error");
            abort();
        }
    }

public:
    using Functor = std::function<void()>;
    EventLoop()
        : _thread_id(std::this_thread::get_id()),
          _eventfd(CreateEventFd()),
          _event_channel(std::make_unique<Channel>(_eventfd, this)),
          _time_wheel(this)
    {
        // 给_eventfd这个描述符添加可读事件监听
        _event_channel->SetReadCallBack(std::bind(&EventLoop::ReadEventFd, this));
        _event_channel->EnableRead();
        //_event_channel->Update();
    }
    // 执行任务，对不在该线程的任务进行压栈
    void RunInLoop(const Functor &cb)
    {
        if (IsInLoop())
        {
            return cb();
        }
        return QueueInLoop(cb);
    }
    // 压栈
    void QueueInLoop(const Functor &cb)
    {
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _task.push_back(cb);
        }
        // 唤醒因为没有事件就绪导致epoll阻塞
        WeakUpEventFd();
    }
    void AssertInLoop()
    {
        assert(_thread_id == std::this_thread::get_id());
    }
    // 判断当前线程是否在EventLoop同一线程
    bool IsInLoop()
    {
        return _thread_id == std::this_thread::get_id();
    }
    // 进行修改添加事件监控
    void UpdataEvent(Channel *channel)
    {
        _poller.UpdateEvent(channel);
    }
    // 删除事件监控
    void RemoveEvent(Channel *channel)
    {
        _poller.RemoveEvent(channel);
    }
    // 启动事件监控
    void Start()
    {
        while(1)
        {
            // 1.启动监控
            std::vector<Channel *> actives;
            _poller.Poll(actives);
            // 2.处理事件
            for (auto &e : actives)
            {
                e->HandleEvent();
            }
            // 3.执行任务
            RunAllTask();
        }
    }
    // 添加定时任务
    void TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &task)
    {
        _time_wheel.TimerAdd(id, timeout, task);
    }
    // 取消定时任务
    void TimerCancel(uint64_t id)
    {
        _time_wheel.TimerCancel(id);
    }
    void TimerRefresh(uint64_t id)
    {
        _time_wheel.TimerRefresh(id);
    }
    bool HasTimer(uint64_t id)
    {
        return _time_wheel.HasTimer(id);
    }

public:
    // 处理栈中的所有任务
    void RunAllTask()
    {
        std::vector<Functor> functor;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            functor.swap(_task);
        }
        for (auto &task : functor)
        {
            task();
        }
    }

private:
    std::thread::id _thread_id; // 判断是否在同一线程
    int _eventfd;               // 进行事件通知
    std::vector<Functor> _task; // 如果当前事件不是在EventLoop线程中处理的就放入任务栈中，最后在一起进行处理
    Poller _poller;             // 对事件进行监控
    std::unique_ptr<Channel> _event_channel;
    std::mutex _mutex;
    TimerWheel _time_wheel;
};

class Any
{
public:
    Any():_content(nullptr) {}
    template <class T>
    Any(const T &val) : _content(new placeholder<T>(val))
    {
    }
    Any(const Any &other) : _content(other._content ? other._content->clone() : nullptr) {}
    ~Any() { delete _content; }

    Any Swap(Any &other)
    {
        std::swap(_content, other._content);
        return *this;
    }

    // 返回子类保存数据的指针
    template <class T>
    T *Get()
    {
        assert(typeid(T) == _content->type());
        return &((placeholder<T> *)_content)->_val;
    }

    Any &operator=(const Any &other)
    {
        Any(other).Swap(*this);
        return *this;
    }
    template <class T>
    Any &operator=(const T &val)
    {
        Any(val).Swap(*this);
        return *this;
    }

private:
    class holder
    {
    public:
        holder() {}
        virtual ~holder() {}
        virtual const std::type_info &type() = 0; // 获取类型
        virtual holder *clone() = 0;              // 克隆
    };
    template <class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T &val) : _val(val) {}
        ~placeholder() {}
        const std::type_info &type() { return typeid(T); }
        holder *clone() { return new placeholder<T>(_val); }

    public:
        T _val;
    };
    holder *_content;
};

typedef enum
{
    DISCONNECTED, // 连接已经关闭
    CONNECTING,   // 连接已建立，但还未初始化各个回调处理
    CONNECTED,    // 连接已经建立成功，可以正常通信
    DISCONNECTING // 连接待关闭的状态
} ConnStatu;

class Connection;
using ConnectionPtr = std::shared_ptr<Connection>;
class Connection : public std::enable_shared_from_this<Connection>
{
public:
    using ConnectCallBack = std::function<void(const ConnectionPtr &)>;           // 对连接建立成功的回调
    using MessageCallBack = std::function<void(const ConnectionPtr &, Buffer *)>; // 对消息处理的回调
    using CloseCallBack = std::function<void(const ConnectionPtr &)>;             // 对连接建立关闭的回调
    using EventCallBack = std::function<void(const ConnectionPtr &)>;             // 对任意事件的回调
public:
    Connection(EventLoop *loop, uint64_t conn_id, int sockfd)
        : _conn_id(conn_id),
          _sockfd(sockfd),
          _conn_statu(CONNECTING),
          _enable_inactive_release(false),
          _loop(loop),
          _channel(_sockfd, _loop),
          _socket(_sockfd)
    {
        _channel.SetReadCallBack(std::bind(&Connection::HandleRead, this));
        _channel.SetWriteCallBack(std::bind(&Connection::HandleWrite, this));
        _channel.SetCloseCallBack(std::bind(&Connection::HandleClose, this));
        _channel.SetErrorCallBack(std::bind(&Connection::HandleError, this));
        _channel.SetEventCallBack(std::bind(&Connection::HandleEvent, this));
        _socket.NonBlock();
    }
    ~Connection() { DLOG("RELEASE CONNECTION:%p", this); }
    // 发送数据
    void Send(const char *data, size_t len)
    {
        if (_conn_statu == DISCONNECTED)
            return;
        _loop->RunInLoop(std::bind(&Connection::SendInLoop, this, data, len));
    }
    // 关闭连接,关闭时先检查是否还有数据未接收或发生
    void ShutDown()
    {
        _loop->RunInLoop(std::bind(&Connection::ShutDownInLoop, this));
    }
    // 启动非活跃销毁
    void EnableInactiveRelase(int sec)
    {
        _loop->RunInLoop(std::bind(&Connection::EnableInactiveRelaseInLoop, this, sec));
    }
    // 关闭非活跃销毁
    void CancalInactiveRelase()
    {
        _loop->RunInLoop(std::bind(&Connection::CancalInactiveRelaseInLoop, this));
    }
    // 切换协议
    void Upgrade(const Any &context, const ConnectCallBack &conn, const MessageCallBack &mess,
                 const CloseCallBack &close, const EventCallBack &event)
    {
        _loop->AssertInLoop();
        _loop->RunInLoop(std::bind(&Connection::UpgradeInLoop, this, context, conn, mess, close, event));
    }
    // 设置上下文
    void SetContext(const Any &context)
    {
        _context = context;
    }
    // 获取上下文
    Any *GetContext() { return &_context; }
    // 获取连接id
    int Id() { return _conn_id; }
    // 获取管理的文件描述符
    int Fd() { return _sockfd; }
    // 获取连接状态
    bool Status() { return _conn_statu == CONNECTED; }
    void SetConnectCallBack(const ConnectCallBack &cb) { _conn_call_back = cb; }
    void SetMessageCallBack(const MessageCallBack &cb) { _msg_call_back = cb; }
    void SetCloseCallBack(const CloseCallBack &cb) { _close_call_back = cb; }
    void SetEventCallBack(const EventCallBack &cb) { _event_call_back = cb; }
    void SetSrvCloseCallBack(const CloseCallBack &cb) { _server_close_call_back = cb; }
    // 启动监听
    void Established()
    {
        _loop->RunInLoop(std::bind(&Connection::EstablishedInLoop, this));
    }
    void Release()
    {
        _loop->QueueInLoop(std::bind(&Connection::ReleaseInLoop, this));
    }
private:
    void SendInLoop(const char *data, size_t len)
    {
        // 将数据发送到发送缓冲区
        if (_conn_statu == DISCONNECTED)
            return;
        _out_buffer.EnsureWriteSpace(len);
        _out_buffer.WriteAndPush(data, len);//bug在发送完之后没有将写偏移到正确的位置，导致给客户端发送数据时没有可读数据
        // 检查当前是否启动的可写事件监听
        if (_channel.WriteAble() == false)
        {
            _channel.EnableWrite(); // 启动写事件监听
            //DLOG("启动写事件监控");
        }
    }
    // 连接关闭处理函数
    void ReleaseInLoop()
    {
        auto self = shared_from_this();
        // 1.修改状态
        _conn_statu = DISCONNECTED;
        // 2.取消监控
        _channel.Remove();
        // 3.关闭描述符
        _socket.Close();
        // 4.查看是否存在定时任务，存在即取消
        if (_loop->HasTimer(_conn_id))
        {
            CancalInactiveRelaseInLoop();
        }
        if (_close_call_back)
            _close_call_back(shared_from_this());
        if (_server_close_call_back)
            _server_close_call_back(shared_from_this());
    }
    void ShutDownInLoop()
    {
        // 1.修改连接状态
        _conn_statu = DISCONNECTING;
        // 2.检查接收缓冲区是否存在未处理的数据
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if(_msg_call_back) _msg_call_back(shared_from_this(), &_in_buffer);
        }
        // 3.发送缓冲区还有数据未发送
        if (_out_buffer.ReadAbleSize() > 0)
        {
            if (_channel.WriteAble() == false)
            {
                _channel.EnableWrite();
            }
        }
        // 4.没有数据直接关闭
        if (_out_buffer.ReadAbleSize() == 0)
        {
            Release();
        }
    }
    void EnableInactiveRelaseInLoop(int sec)
    {
        // 1.将标志位置为true
        _enable_inactive_release = true;
        // 2.检查是否存在该定时任务
        //   存在---刷新
        if (_loop->HasTimer(_conn_id))
        {
            return _loop->TimerRefresh(_conn_id);
        }
        // 不存在则添加
        _loop->TimerAdd(_conn_id, sec, std::bind(&Connection::Release, this));
    }
    void CancalInactiveRelaseInLoop()
    {
        _enable_inactive_release = false;
        if (_loop->HasTimer(_conn_id))
            _loop->TimerCancel(_conn_id);
    }
    // 更换协议
    void UpgradeInLoop(const Any &context, const ConnectCallBack &conn, const MessageCallBack &mess,
                       const CloseCallBack &close, const EventCallBack &event)
    {
        _context = context;
        _conn_call_back = conn;
        _msg_call_back = mess;
        _close_call_back = close;
        _event_call_back = event;
    }
    // channel中可读事件的回调函数
    void HandleRead()
    {
        // 通过套接字读取数据
        char buff[65536];
        int n = _socket.RecvNonBlock(buff, 65535);
        //DLOG("RecvNonBlock return: %d", n);
        if (n < 0)
        {
            // 读取异常
            return ShutDownInLoop();
        }
        else if (n == 0)
        {
            // 对端关闭连接
            DLOG("Connection closed by peer");
            return ShutDownInLoop();
        }

        // 将数据放入接收缓冲区
        _in_buffer.WriteAndPush(buff, n);
        if (_in_buffer.ReadAbleSize() > 0)
        {
            // 调用消息处理回调函数
            return _msg_call_back(shared_from_this(), &_in_buffer);
        }
    }
    // channel中可写事件的回调函数
    void HandleWrite()
    {
        ssize_t n = _socket.NonBlockSend(_out_buffer.ReadPosition(), _out_buffer.ReadAbleSize());
        //DLOG("NonBlockSend return: %ld", n); // 添加日志
        if (n < 0)
        {
            // 发送数据失败
            // 如果接收缓冲区还有数据就需要进行处理
            if (_in_buffer.ReadAbleSize() > 0)
            {
                _msg_call_back(shared_from_this(), &_in_buffer);
            }
            return Release(); // 直接关闭连接
        }
        _out_buffer.MoveReadOffset(n);
        if (_out_buffer.ReadAbleSize() == 0)
        {
            // 当发生缓冲区没有数据时将可写事件监控关闭
            _channel.DisableWrite();
            // 如果连接为待关闭状态
            if (_conn_statu == DISCONNECTING)
            {
                return ShutDownInLoop();
            }
        }
    }
    // channel中关闭事件的回调函数
    void HandleClose()
    {
        if (_in_buffer.ReadAbleSize() > 0)
        {
            if(_msg_call_back) _msg_call_back(shared_from_this(), &_in_buffer);
        }
        return Release(); // 直接关闭连接
    }
    // channel中任意事件的回调函数
    void HandleEvent()
    {
        if (_enable_inactive_release == true)
        {
            _loop->TimerRefresh(_conn_id);
        }
        if (_event_call_back)
            _event_call_back(shared_from_this());
    }
    // channel中错误事件的回调函数
    void HandleError()
    {
        return HandleClose();
    }
    // 设置channel的所有回调函数，用于获取连接之后
    void EstablishedInLoop()
    {
        // 1.修改连接状态
        assert(_conn_statu == CONNECTING);
        _conn_statu = CONNECTED;
        // 2.启动可读事件监控
        _channel.EnableRead();
        if (_conn_call_back)
            return _conn_call_back(shared_from_this());
    }

private:
    uint64_t _conn_id; // 连接id
    int _sockfd;
    ConnStatu _conn_statu;         // 连接的状态，例如在关闭连接时可能缓冲区还有数据没有处理
    bool _enable_inactive_release; // 是否启动非活跃关闭
    Socket _socket;
    EventLoop *_loop;
    Channel _channel; // 事件处理模块
    Buffer _in_buffer;
    Buffer _out_buffer;
    Any _context; // 对上下文的管理

    ConnectCallBack _conn_call_back;
    MessageCallBack _msg_call_back;
    CloseCallBack _close_call_back;
    EventCallBack _event_call_back;
    CloseCallBack _server_close_call_back;
};

//只对监听套接字进行管理
class Acceptor
{
public:
    using AcceptCallBack = std::function<void(int)>;
    Acceptor(EventLoop* loop,int port):_loop(loop),_socket(CreateServer(port)),_channel(_socket.Fd(),_loop)
    {
        //DLOG("构造");
        _channel.SetReadCallBack(std::bind(&Acceptor::HandleRead,this));
    }
    void SetAcceptCallBack(const AcceptCallBack& cb){ _accept_callback=cb; }
    void Listen() { _channel.EnableRead(); }
private:
    //读事件触发的回调函数，获取新连接，使用_accept_callback对连接进行处理
    void HandleRead()
    {
        //获取新连接
        //DLOG("-------------");
        int newfd = _socket.Accept();
        if(newfd<0)
        {
            DLOG("获取新连接失败");
            return;
        }
        if(_accept_callback) _accept_callback(newfd);
    }
    int CreateServer(int port)
    {
        bool ret = _socket.CreateServer(port);
        assert(ret==true);
        DLOG("%d",_socket.Fd());
        return _socket.Fd();
    }
private:
    Socket _socket;
    EventLoop* _loop;
    Channel _channel;
    AcceptCallBack _accept_callback;
};

//一个线程对应一个EventLoop,注意：必须是先有线程，然后通过线程实例化出loop对象
class LoopThread
{
public:
    //创建线程
    LoopThread():_thread(std::thread(&LoopThread::ThreadEntry,this))
    {}
    //返回当前线程关联的EventLoop指针
    EventLoop* GetLoop()
    {
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _cond.wait(lock,[&](){return _loop != nullptr;});
            loop=_loop;
        }
        return loop;
    }
private:
    //实例化loop对象，并且启动loop
    void ThreadEntry()
    {
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(_mutex);
            _loop=&loop;
            //唤醒正在等待的线程
            _cond.notify_all();
        }
        _loop->Start();
    }
private:
    //防止在线程未实例化loop对象之前获取loop对象
    std::mutex _mutex;
    std::condition_variable _cond;
    std::thread _thread;
    EventLoop* _loop=nullptr;
};

class LoopThreadPool
{
public:
    LoopThreadPool(EventLoop* loop):_thread_count(0),_next_idx(0),_baseloop(loop)
    {}
    void SetThreadCount(int count)
    {
        _thread_count=count;
    }
    void Create()
    {
        if(_thread_count>0)
        {
            _threads.resize(_thread_count);
            _loops.resize(_thread_count);
            for(int i = 0;i<_thread_count;i++)
            {
                _threads[i]=new LoopThread();
                _loops[i]=_threads[i]->GetLoop();
            }
        }
    }
    EventLoop* NextLoop()
    {
        if(_thread_count==0)
        {
            return _baseloop;
        }
        _next_idx=(_next_idx+1)%_thread_count;
        return _loops[_next_idx];
    }
private:
    int _thread_count;//线程数
    int _next_idx;//轮转
    EventLoop* _baseloop;
    std::vector<LoopThread*> _threads;
    std::vector<EventLoop*> _loops;
};

class TcpServer
{
public:
    using ConnectCallBack = std::function<void(const ConnectionPtr &)>;           // 对连接建立成功的回调
    using MessageCallBack = std::function<void(const ConnectionPtr &, Buffer *)>; // 对消息处理的回调
    using CloseCallBack = std::function<void(const ConnectionPtr &)>;             // 对连接建立关闭的回调
    using EventCallBack = std::function<void(const ConnectionPtr &)>;             // 对任意事件的回调
    using Functor = std::function<void()>;
public:
    TcpServer(const int port)
    :_next_id(0),
    _port(port),
    _enable_inactive_release(false),
    _acceptor(&_baseloop,_port),
    _pool(&_baseloop)
    {
        _acceptor.SetAcceptCallBack(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1));
        _acceptor.Listen();
    }
    //设置线程数
    void SetThreadLoopCount(int num) { _pool.SetThreadCount(num); }
    //启动服务器
    void Start()
    {
        _pool.Create();
        _baseloop.Start();
    }
    //添加定时任务
    void RunAfter(const Functor& fun,int delay)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,fun,delay));
    }
    //启动非活跃销毁
    void EnableInactiveRelease(int timeout)
    {
        _timeout=timeout;
        _enable_inactive_release = true;
    }
    void SetConnectCallBack(const ConnectCallBack &cb) { _conn_call_back = cb; }
    void SetMessageCallBack(const MessageCallBack &cb) { _msg_call_back = cb; }
    void SetCloseCallBack(const CloseCallBack &cb) { _close_call_back = cb; }
    void SetEventCallBack(const EventCallBack &cb) { _event_call_back = cb; }
private:
    //获取一个新连接并添加进管理模块
    void NewConnection(int fd)
    {
        _next_id++;
        ConnectionPtr conn(new Connection(_pool.NextLoop(),_next_id,fd));
        conn->SetSrvCloseCallBack(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
        conn->SetCloseCallBack(_close_call_back);
        conn->SetConnectCallBack(_conn_call_back);
        conn->SetMessageCallBack(_msg_call_back);
        conn->SetEventCallBack(_event_call_back);
        //启动非活跃销毁
        if(_enable_inactive_release) conn->EnableInactiveRelase(_timeout);
        conn->Established();
        _conns.insert(std::make_pair(_next_id,conn));
    }
    void RemoveConnectionInLoop(const ConnectionPtr &conn)
    {
        int id = conn->Id();
        auto it = _conns.find(id);
        if(it==_conns.end())
        {
            return;
        }
        _conns.erase(it);
        //DLOG("relese connection %p",conn.get());
    }
    //删除一个连接
    void RemoveConnection(const ConnectionPtr &conn)
    {
        _baseloop.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
    }
    void RunAfterInLoop(const Functor& fun,int delay)
    {
        _next_id++;
        _baseloop.TimerAdd(_next_id,delay,fun);
    }
private:
    int _next_id;
    int _port;
    EventLoop _baseloop;
    int _timeout;//非活跃超时时间
    bool _enable_inactive_release; // 是否启动非活跃关闭
    Acceptor _acceptor;
    LoopThreadPool _pool;
    std::unordered_map<int,ConnectionPtr> _conns;

    ConnectCallBack _conn_call_back;
    MessageCallBack _msg_call_back;
    CloseCallBack _close_call_back;
    EventCallBack _event_call_back;
};

// 移除事件监控
void Channel::Remove()
{
    _eventloop->RemoveEvent(this);
}
void Channel::Update()
{
    _eventloop->UpdataEvent(this);
}
void TimerWheel::TimerAdd(uint64_t id, uint32_t timeout, const TaskFunc &task)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop, this, id, timeout, task));
}
// 刷新/延长超时时间
void TimerWheel::TimerRefresh(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop, this, id));
}
// 取消
void TimerWheel::TimerCancel(uint64_t id)
{
    _loop->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop, this, id));
}

class NetWork {
    public:
        NetWork() {
            DLOG("SIGPIPE INIT");
            signal(SIGPIPE, SIG_IGN);
        }
};
static NetWork nw;
#endif